using System;
using System.IO;
using System.Collections.Generic;
using System.Text;
using System.Diagnostics;
using System.Threading;
using System.Net;
using System.Net.Sockets;
using Microsoft.BizTalk.Message.Interop;
using Microsoft.BizTalk.TransportProxy.Interop;
using Microsoft.Samples.BizTalk.Adapter.Tcp;

namespace Microsoft.Samples.BizTalk.Adapter.Tcp.Send
{
    /// <summary>
    ///     This worker thread transmits data from BizTalk to the remote endpoint.
    /// </summary>
    /// <author>
    ///     Paul Brown
    /// </author>
    public class TransmitWorkerThread : WorkerThread
    {
        #region Private Instance Members

        private SendAdapter _sendAdapter;   // The send adapter

        #endregion

        #region Constructor

        /// <summary>
        ///     This is constructor initializes the instance.
        /// </summary>
        public TransmitWorkerThread(IAdapter adapter)
            : base(adapter)
        {
            // Get thread name prefix
            this.ThreadNamePrefix = TraceHelper.ResourceManager.GetString(ResourceConstants.SendTransmitWorkerThreadName);

            // Get reference to send adapter
            this._sendAdapter = (SendAdapter)adapter;
        }

        #endregion

        #region Public Instance Properties

        /// <summary>
        ///     This property returns a reference to the send adapter.
        /// </summary>
        public SendAdapter SendAdapter
        {
            get { return (this._sendAdapter); }
        }

        #endregion

        #region Private Instance Methods

        /// <summary>
        ///     This method creates a BizTalk message for submission into the batch.
        /// </summary>
        private IBaseMessage CreateMessage(Stream stream, IBaseMessageFactory messageFactory, TransmitWorkItem workItem)
        {
            // Create message part to contain the body of the message
            IBaseMessagePart part = messageFactory.CreateMessagePart();
            part.Data = stream;

            // Create message
            IBaseMessage message = messageFactory.CreateMessage();
            message.AddPart("body", part, true);

            // Return message
            return (message);
        }

        /// <summary>
        ///     This method sends the data associated with the work item to the remote endpoint for one-way send locations.
        /// </summary>
        private void SendData(TransmitWorkItem workItem)
        {
            // Set up remote endpoint
            IPHostEntry host = Dns.GetHostEntry(workItem.SendLocation.EndpointAddress);

            #region JPLUMMER IPV6 
            //IPEndPoint remoteEndpoint = new IPEndPoint(host.AddressList[0], workItem.SendLocation.EndpointPort);
            IPEndPoint remoteEndpoint = null; ;
            foreach (IPAddress addr in host.AddressList)
            {
                if (addr.AddressFamily == AddressFamily.InterNetwork)
                {
                    remoteEndpoint = new IPEndPoint(addr, workItem.SendLocation.EndpointPort);
                    break;
                }
            }
            #endregion

            // Create socket, bind and then connect
            Socket endpointSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            IPEndPoint localEndpoint = new IPEndPoint(IPAddress.Any, 0);
            
            

            endpointSocket.Bind(localEndpoint);
            endpointSocket.Connect(remoteEndpoint);

            #region Broken Code -  JPLUMMER 14/6/06
            // Send data in message
            //long totalBytesSent = SocketIO.SendData(this.Adapter,
            //                                        endpointSocket,
            //                                        new Stream[] { workItem.SendLocation.Message.BodyPart.Data },
            //                                        workItem.SendLocation.FrameStart,
            //                                        workItem.SendLocation.FrameEnd,
            //                                        workItem.SendLocation.DataEncoding,
            //                                        workItem.SendLocation.SendDataBufferSize);
            #endregion

            //check if source stream can seek - JPLUMMER 14/6/06
            Stream sourceStream = workItem.SendLocation.Message.BodyPart.GetOriginalDataStream();

            if (!sourceStream.CanSeek)
            {
                // Create a virtual (seekable) stream
                SeekableReadOnlyStream seekableStream = new SeekableReadOnlyStream(sourceStream);

                // Set new stream for the body part data of the input message. This new stream will then used for further processing.
                // We need to do this because input stream may not support seeking, so we wrap it with a seekable stream.
                workItem.SendLocation.Message.BodyPart.Data = seekableStream;
            }

            long totalBytesSent = SocketIO.SendData(this.Adapter,
                                                    endpointSocket,
                                                    new Stream[] { workItem.SendLocation.Message.BodyPart.Data },
                                                    workItem.SendLocation.FrameStart,
                                                    workItem.SendLocation.FrameEnd,
                                                    workItem.SendLocation.DataEncoding,
                                                    workItem.SendLocation.SendDataBufferSize);

            // If we are duplex sending, get response to see how much data was sent and check against our own total
            if (workItem.SendLocation.SupportsDuplex)
            {
                // Receive 8 bytes from socket
                byte[] buffer = new byte[8];
                endpointSocket.Receive(buffer, buffer.Length, SocketFlags.None);

                // Convert result into a long
                long totalBytesSentReceived = IPAddress.NetworkToHostOrder(BitConverter.ToInt64(buffer, 0));

                if (totalBytesSent != totalBytesSentReceived)
                {
                    // Close socket
                    endpointSocket.Shutdown(SocketShutdown.Send);
                    endpointSocket.Close(workItem.SendLocation.CloseTimeout);

                    // Not equal - duplex worker thread on receive side didn't transmit same number of bytes back to initiator
                    Exception e = new Exception(TraceHelper.FormatEntry(ResourceConstants.SendDuplexSendBytesNotEqual, this.Name, totalBytesSent, workItem.SendLocation.EndpointAddress, workItem.SendLocation.EndpointPort, totalBytesSentReceived));
                    TraceHelper.WriteRawEntry(TraceLevel.Error, e.Message);
                    throw (e);
                }
            }
            else
            {
                // Close socket
                endpointSocket.Shutdown(SocketShutdown.Send);
                endpointSocket.Close(workItem.SendLocation.CloseTimeout);
            }
        }

        /// <summary>
        ///     This method sends the data associated with the work item to the remote endpoint and receives
        ///     a response if it is a solicit-response send location.
        /// </summary>
        private Stream[] SendDataRecvResponse(TransmitWorkItem workItem, out bool isLastMessageNotComplete)
        {
            // Set up remote endpoint
            IPHostEntry host = Dns.GetHostEntry(workItem.SendLocation.EndpointAddress);
            #region JPLUMMER IPV6
            //IPEndPoint remoteEndpoint = new IPEndPoint(host.AddressList[0], workItem.SendLocation.EndpointPort);
            IPEndPoint remoteEndpoint = null; ;
            foreach (IPAddress addr in host.AddressList)
            {
                if (addr.AddressFamily == AddressFamily.InterNetwork)
                {
                    remoteEndpoint = new IPEndPoint(addr, workItem.SendLocation.EndpointPort);
                    break;
                }
            }
            #endregion

            // Create socket, bind and then connect
            Socket endpointSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            IPEndPoint localEndpoint = new IPEndPoint(IPAddress.Any, 0);
            endpointSocket.Bind(localEndpoint);
            endpointSocket.Connect(remoteEndpoint);

            #region JPLUMMER 6th October 2006
            // Send data in message
            //SocketIO.SendData(this.Adapter,
            //                  endpointSocket,
            //                  new Stream[] { workItem.SendLocation.Message.BodyPart.Data },
            //                  workItem.SendLocation.FrameStart,
            //                  workItem.SendLocation.FrameEnd,
            //                  workItem.SendLocation.DataEncoding,
            //                  workItem.SendLocation.SendDataBufferSize);

            
           Stream sourceStream = workItem.SendLocation.Message.BodyPart.GetOriginalDataStream();

            if (!sourceStream.CanSeek)
            {
                // Create a virtual (seekable) stream
                SeekableReadOnlyStream seekableStream = new SeekableReadOnlyStream(sourceStream);

                // Set new stream for the body part data of the input message. This new stream will then used for further processing.
                // We need to do this because input stream may not support seeking, so we wrap it with a seekable stream.
                workItem.SendLocation.Message.BodyPart.Data = seekableStream;
            }

            long totalBytesSent = SocketIO.SendData(this.Adapter,
                                                    endpointSocket,
                                                    new Stream[] { workItem.SendLocation.Message.BodyPart.Data },
                                                    workItem.SendLocation.FrameStart,
                                                    workItem.SendLocation.FrameEnd,
                                                    workItem.SendLocation.DataEncoding,
                                                    workItem.SendLocation.SendDataBufferSize);






            #endregion







            // If we are solicit-response, receive response
            bool notComplete;
            Stream[] streams = null;
            if (workItem.SendLocation.IsTwoWay)
            {
                streams = SocketIO.ReceiveData(this.Adapter,
                                               endpointSocket,
                                               workItem.SendLocation.FrameStart,
                                               workItem.SendLocation.FrameEnd,
                                               workItem.SendLocation.DataEncoding,
                                               workItem.SendLocation.ReceiveDataBufferSize,
                                               false,
                                               out notComplete);
            }
            else
            {
                // Assign flag default value
                notComplete = false;
            }

            // Close socket
            endpointSocket.Shutdown(SocketShutdown.Both);
            endpointSocket.Close(workItem.SendLocation.CloseTimeout);

            // Return streams if any
            isLastMessageNotComplete = notComplete;
            return (streams);
        }

        #endregion

        #region Abstract WorkerThread Members

        #region Public Instance Properties

        /// <summary>
        ///     This property returns the type of a BatchWorkItem which is compatible with this worker thread.
        /// </summary>
        public override Type WorkItemType
        {
            get { return (typeof(TransmitWorkItem)); }
        }

        #endregion

        #region Protected Instance Methods

        /// <summary>
        ///     This method is the delegate that runs the main worker thread.
        /// </summary>
        protected override void Thread()
        {
            // Variables
            bool                threadInterrupted = false;
            IBaseMessage        message = null;
            TransmitWorkItem    workItem = null;
            BatchWorkItem       batchWorkItem = null;
            BatchMessage        batchMessage = null;

            try
            {
                // Loop until we are interrupted
                while (true)
                {
                    // Loop while we have work items
                    while (this.CurrentWorkLoad > 0)
                    {
                        // Dequeue item
                        workItem = (TransmitWorkItem)this.DequeueWork();

                        if (this.Adapter.TraceSwitch.TraceVerbose)
                            TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendDequeuedTransmitWork, workItem.SendLocation.Uri, this.Name);

                        try
                        {
                            // Deliver message to remote endpoint
                            if (workItem.SendLocation.IsTwoWay)
                            {
                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendSendingDataSolicitResponse, workItem.SendLocation.EndpointAddress, workItem.SendLocation.EndpointPort, this.Name);

                                // Two way
                                bool isLastMessageNotComplete;
                                Stream[] streams = this.SendDataRecvResponse(workItem, out isLastMessageNotComplete);

                                // Any responses?
                                if (streams != null && streams.Length > 0)
                                {
                                    // One 1 response and it is not complete!
                                    if (isLastMessageNotComplete && streams.Length == 0)
                                    {
                                        // Create exception
                                        string errorMessage = TraceHelper.FormatEntry(ResourceConstants.SendPartialResponseReceived, workItem.SendLocation.Uri, this.Name);
                                        TraceHelper.WriteRawEntry(TraceLevel.Error, errorMessage);
                                        Exception msgEx = new Exception(errorMessage);

                                        // Create message from partial data
                                        message = this.CreateMessage(streams[0], this.Adapter.TransportProxy.GetMessageFactory(), workItem);

                                        // Suspend message
                                        batchWorkItem = new BatchWorkItem(workItem.SendLocation);
                                        batchMessage = new BatchMessage(message, msgEx);
                                        batchMessage.BatchOperationType = BatchOperationType.MoveToSuspendQ;
                                        batchWorkItem.BatchMessages.Add(batchMessage);

                                        // Dispatch work
                                        this.Adapter.Dispatcher.AddWorkItem(batchWorkItem);
                                    }
                                    else
                                    {
                                        // More than 1 response not supported
                                        if (streams.Length > 1)
                                        {
                                            // Discarding other responses
                                            if (this.Adapter.TraceSwitch.TraceVerbose)
                                                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendDiscardingOtherResponses, workItem.SendLocation.Uri, this.Name);
                                        }

                                        if (this.Adapter.TraceSwitch.TraceVerbose)
                                            TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendReceivedResponse, workItem.SendLocation.Uri, this.Name);

                                        // Must have been two way and we have a response
                                        message = this.CreateMessage(streams[0], this.Adapter.TransportProxy.GetMessageFactory(), workItem);

                                        // Submit response and delete of solicit message in same batch
                                        batchWorkItem = new BatchWorkItem(workItem.SendLocation);

                                        batchMessage = new BatchMessage(message, workItem.SendLocation.Message);
                                        batchMessage.BatchOperationType = BatchOperationType.Submit;
                                        batchWorkItem.BatchMessages.Add(batchMessage);

                                        batchMessage = new BatchMessage(workItem.SendLocation.Message);
                                        batchMessage.BatchOperationType = BatchOperationType.Delete;
                                        batchWorkItem.BatchMessages.Add(batchMessage);

                                        // Dispatch work
                                        this.Adapter.Dispatcher.AddWorkItem(batchWorkItem);
                                    }
                                }
                                else
                                {
                                    // No response, throw error to cause this thread to retry
                                    throw (new Exception(TraceHelper.FormatEntry(ResourceConstants.SendReceivedNoResponse, workItem.SendLocation.Uri, this.Name)));
                                }
                            }
                            else
                            {
                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendSendingDataSolicit, workItem.SendLocation.EndpointAddress, workItem.SendLocation.EndpointPort, this.Name);

                                // One way
                                this.SendData(workItem);

                                // Delete message in BizTalk
                                batchWorkItem = new BatchWorkItem(workItem.SendLocation);
                                batchMessage = new BatchMessage(workItem.SendLocation.Message);
                                batchMessage.BatchOperationType = BatchOperationType.Delete;
                                batchWorkItem.BatchMessages.Add(batchMessage);

                                // Dispatch work
                                this.Adapter.Dispatcher.AddWorkItem(batchWorkItem);
                            }
                        }
                        catch (ThreadInterruptedException)
                        {
                            // Asked to shut down, set flag and wait until this batch is done
                            threadInterrupted = true;
                        }
                        catch (Exception e)
                        {
                            // Create exception
                            string errorMessage = TraceHelper.FormatEntry(ResourceConstants.SendTransmitFailure, workItem.SendLocation.EndpointAddress, workItem.SendLocation.EndpointPort, e.Message);
                            TraceHelper.WriteRawEntry(TraceLevel.Error, errorMessage);
                            Exception msgEx = new Exception(errorMessage, e);

                            // Resubmit or move to next transport
                            SystemMessageContext context = new SystemMessageContext(workItem.SendLocation.Message.Context);

                            if (context != null && context.RetryCount > 0)
                            {
                                // Resubmit
                                batchWorkItem = new BatchWorkItem(workItem.SendLocation);
                                batchMessage = new BatchMessage(workItem.SendLocation.Message, msgEx);
                                batchMessage.BatchOperationType = BatchOperationType.Resubmit;
                                batchWorkItem.BatchMessages.Add(batchMessage);
                            }
                            else
                            {
                                // Move to next transport
                                batchWorkItem = new BatchWorkItem(workItem.SendLocation);
                                batchMessage = new BatchMessage(workItem.SendLocation.Message, msgEx);
                                batchMessage.BatchOperationType = BatchOperationType.MoveToNextTransport;
                                batchWorkItem.BatchMessages.Add(batchMessage);
                            }

                            // Dispatch work
                            this.Adapter.Dispatcher.AddWorkItem(batchWorkItem);
                        }

                        // Update work item
                        workItem.Status = WorkItemStatusTypes.Complete;
                        workItem = null;
                    }

                    // Was the thread interrupted?
                    if (threadInterrupted)
                    {
                        // Throw error again to end thread
                        throw (new ThreadInterruptedException());
                    }

                    if (this.Adapter.TraceSwitch.TraceVerbose)
                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.WorkerThreadWaitingForWork, this.Name);

                    // Wait for signal that work has arrived
                    this.WaitForWork();
                }
            }
            catch (ThreadInterruptedException)
            {
                // Asked to shut down
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.WorkerThreadInterrupted, this.Name);

                // If we have any in-flight work still, save it off
                if (workItem != null)
                {
                    workItem.Save();
                }

                while (this.CurrentWorkLoad > 0)
                {
                    workItem = (TransmitWorkItem)this.DequeueWork();
                    workItem.Save();
                }
            }
            catch (Exception e)
            {
                TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.WorkerThreadGeneralError, this.Name, e.Message);

                // If we have an outstanding work item, push it back onto queue before restarting
                if (workItem != null)
                {
                    this.PushWorkOnQueue(workItem);
                }
            }
            finally
            {
                // Fire terminate event
                this.OnThreadTerminated(new EventArgs());
            }
        }

        #endregion

        #endregion
    }
}
